package com.example.batch;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * @author lbq
 * @date 2022/3/14 22:13
 */
public class WordCount {

    private final static String INPUT_PATH = "data/input/text.txt";
    private final static String OUTPUT_PATH = "data/output/result.csv";

    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        final DataSet<String> text = env.readTextFile(INPUT_PATH);
        final AggregateOperator<Tuple2<String, Integer>> sum = text.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>)(line, collector) -> {
            final String[] words = line.split("\\s+");
            for (String word : words) {
                collector.collect(new Tuple2<>(word, 1));
            }
        }).groupBy(0).sum(1);
        sum.writeAsCsv(OUTPUT_PATH, "\n", ",").setParallelism(1);
        env.execute(WordCount.class.getName());
    }
}
